package rxjava.operators;


import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.Observer;
import io.reactivex.disposables.Disposable;

/**
 * User: Administrator
 * Date: 2021/3/23
 * Time: 10:56
 * Description:
 */
public class CreateDemo {
    public static void main(String[] args) {
        Observable.create((ObservableOnSubscribe<Integer>) observableEmitter -> {
            observableEmitter.onNext(1);
            observableEmitter.onNext(2);
            observableEmitter.onNext(3);
            observableEmitter.onNext(4);
            observableEmitter.onComplete();
        }).subscribe(new Observer<Integer>() {
            private int i;
            //disposable为true时  观察者停止接受事件
            private Disposable disposable;

            @Override
            public void onSubscribe(Disposable disposable) {
                System.out.println("Observer  onSubscribe  ====");
                this.disposable = disposable;
            }

            @Override
            public void onNext(Integer a) {
                i ++;
                System.out.println("Observer onNext ===" + a);
                if (i > 3) {
                    this.disposable.dispose();
                    System.out.println(this.disposable.isDisposed());
                }
            }

            @Override
            public void onError(Throwable throwable) {
                System.out.println("Observer onError");
            }

            @Override
            public void onComplete() {
                System.out.println("Observer onComplete");
            }
        });
    }
}
